package me.tuyou.stream;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class ProcessWindowTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        DataStreamSource<Tuple2<String, Long>> input = env.fromElements(
                new Tuple2<>("BMW", 2L),
                new Tuple2<>("BMW", 2L),
                new Tuple2<>("Tesla", 3L),
                new Tuple2<>("Tesla", 4L)
        );

        input.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Tuple2<String, Long>>() {
                    @Override
                    public long extractAscendingTimestamp(Tuple2<String, Long> element) {
                        return element.f1;
                    }
                })
                .keyBy(t -> t.f0)
                .timeWindow(Time.seconds(1))
                .process(new MyProcessWindowFunction())
                .print();
        env.execute();
    }

    static class MyProcessWindowFunction extends ProcessWindowFunction<Tuple2<String, Long>, String, String, TimeWindow> {
        @Override
        public void process(String s, Context context, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
            int count = 0;
            for (Tuple2<String, Long> in : input) {
                count++;
            }
            out.collect("窗口信息：" + context.window() + ", 元素数量：" + count);
        }
    }
}
